package com.yi.wood.kafka.msg;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * Created by moyi on 2019/2/20 10:34
 *
 * @author moyi
 *
 * 服务器世界聊天
 */
@Component
public class ImKafkaMsg implements Msg{

    @Resource
    private SendMsgCallBack sendMsgCallBack;

    @Autowired
    private KafkaTemplate kafkaTemplate;


    @Override
    public void send(String msg) {
        ListenableFuture future =  kafkaTemplate.send(MSG_TOPIC, msg);
        future.addCallback(sendMsgCallBack);
    }

    @KafkaListener(topics = MSG_TOPIC,groupId = MSG_GROUP_ID)
    public void listener(ConsumerRecord<?, ?> record){
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            System.out.println(">>"+message);
        }
    }
}
